use std::{pin, thread};
use std::{pin::pin, sync::mpsc, time::Duration};
use trpl::{Either, Html, ReceiverStream, Stream, StreamExt};

pub fn chain_demo() {
    let (tx, rx) = mpsc::channel();

    // 多生产者
    let tx1 = tx.clone();
    // thread::spawn(move || {
    //     let val = "hi111".to_string();
    //     tx.send(val).unwrap();
    // });

    // let received = rx.recv().unwrap();
    // println!("Got: {}", received)

    thread::spawn(move || {
        let arr = vec![
            "tx1 hi".to_string(),
            "tx1 from".to_string(),
            "tx1 the".to_string(),
            "tx1 thread".to_string(),
        ];
        // 使用into_iter转移所有权
        arr.into_iter().for_each(|ele| {
            tx1.send(ele).unwrap();
            thread::sleep(Duration::from_secs(1));
        });
    });

    thread::spawn(move || {
        let arr = vec![
            "tx more".to_string(),
            "tx messages".to_string(),
            "tx for".to_string(),
            "tx you".to_string(),
        ];
        arr.into_iter().for_each(|ele| {
            tx.send(ele).unwrap();
            thread::sleep(Duration::from_secs(1));
        });
    });

    for ele in rx {
        println!("Got: {}", ele)
    }
}

pub async fn page_title(url: &str) -> (&str, Option<String>) {
    let res = trpl::get(url).await;
    let res_txt = res.text().await;
    let title = Html::parse(&res_txt)
        .select_first("title")
        .map(|ele| ele.inner_html());
    (url, title)
}

// pub fn one_url() {
//     let url = "https://tieba.baidu.com/".to_string();
//     trpl::run(async {
//         match page_title(&url).await {
//             Some(title) => println!("url: {} title: {}", url, title),
//             None => println!("url: {} title: None", url),
//         }
//     });
// }

pub fn two_url() {
    let url1 = "https://www.baidu.com/".to_string();
    let url2 = "https://tieba.baidu.com/".to_string();
    trpl::run(async {
        let f1 = page_title(&url1);
        let f2 = page_title(&url2);
        let (utl, title) = match trpl::race(f1, f2).await {
            Either::Left(left) => left,
            Either::Right(right) => right,
        };
        println!("url:{}", utl);
        match title {
            Some(title) => println!("title:{}", title),
            None => println!("title:None"),
        }
    })
}

pub fn spawn_demo() {
    trpl::run(async {
        let handle = trpl::spawn_task(async {
            for i in 1..10 {
                println!("first task:{i}")
            }
            trpl::sleep(Duration::from_millis(500)).await;
        });

        for i in 1..5 {
            println!("second task:{i}");
            trpl::sleep(Duration::from_millis(500)).await;
        }

        handle.await.unwrap();
    })
}

pub fn spawn_v2() {
    println!("start");
    trpl::run(async {
        let f1 = async {
            for i in 1..10 {
                println!("first task:{i}")
            }
            trpl::sleep(Duration::from_millis(500)).await;
        };
        let f2 = async {
            for i in 1..5 {
                println!("second task:{i}")
            }
            trpl::sleep(Duration::from_millis(500)).await;
        };

        trpl::join(f2, f1).await;
    });
    println!("end");
}

pub fn channel_v2() {
    trpl::run(async {
        let (tx, mut rx) = trpl::channel();
        // let vals = vec![
        //     String::from("hi"),
        //     String::from("from"),
        //     String::from("the"),
        //     String::from("future"),
        // ];
        // for ele in vals {
        //     tx.send(ele).unwrap();
        //     trpl::sleep(Duration::from_millis(500)).await;
        // }

        // while let Some(value) = rx.recv().await {
        //     println!("Got: {}", value)
        // }

        let f1 = async move {
            let vals = vec![
                String::from("hi"),
                String::from("from"),
                String::from("the"),
                String::from("future"),
            ];
            for ele in vals {
                tx.send(ele).unwrap();
                trpl::sleep(Duration::from_millis(500)).await;
            }
        };
        let f2 = async {
            while let Some(val) = rx.recv().await {
                println!("Got: {}", val)
            }
        };
        trpl::join(f1, f2).await;
    })
}

pub fn stream() {
    trpl::run(async {
        // let arr = vec![1, 2, 3, 4, 5];
        // let mut stream = trpl::stream_from_iter(arr.iter().map(|ele| ele * 2));
        // while let Some(val) = stream.next().await {
        //     println!("Got: {val}")
        // }

        // let mut msgs = get_msg();
        // while let Some(val) = msgs.next().await {
        //     println!("Got: {val}")
        // }

        // 具有发送延时的消息流
        // let mut msgs = pin!(get_msg().timeout(Duration::from_millis(200)));
        // while let Some(val) = msgs.next().await {
        //     match val {
        //         Ok(val) => println!("Got: {val}"),
        //         Err(e) => println!("Timeout:{e}"),
        //     }
        // }

        // 合并流
        let msgs = get_msg().timeout(Duration::from_millis(200));
        let intervals = get_intervals()
        .map(|ele| format!("Iterval:{ele}"))
        //限流 限制函数被调用的频率
        .throttle(Duration::from_millis(100))
        .timeout(Duration::from_secs(10));
        let merged = msgs.merge(intervals).take(20);
        let mut stream = pin!(merged);
        while let Some(item) = stream.next().await {
            match item {
                Ok(msg) => println!("{msg}"),
                Err(err) => println!("{err}"),
            }
        }
    })
}

pub fn get_msg() -> impl Stream<Item = String> {
    let (tx, rx) = trpl::channel();
    // for ele in msgs {
    //     tx.send(format!("msg:'{ele}'")).unwrap();
    // }
    trpl::spawn_task(async move {
        let msgs = ["a", "b", "c", "d", "e", "f", "g", "h", "i", "j"];
        for (index, ele) in msgs.into_iter().enumerate() {
            let time_to_sleep = if index % 2 == 0 { 100 } else { 300 };
            trpl::sleep(Duration::from_millis(time_to_sleep)).await;
            match  tx.send(format!("msg:'{ele}'")){
                Ok(()) => {},
                Err(err) => {
                    println!("get_msg_err:{:?}", err);
                    break;
                },
            }
            
        }
    });
    ReceiverStream::new(rx)
}

fn get_intervals() -> impl Stream<Item = u32>{
    let (tx,rx) = trpl::channel();
    trpl::spawn_task(async move{
        let mut count = 0;
        loop {
            trpl::sleep(Duration::from_millis(1)).await;
            count += 1;
            match tx.send(count) {
                Ok(())=>{},
                Err(e)=>{
                    println!("get_intervals_err:{:?}",e);
                    break;
                }
            }
        }
    });
    ReceiverStream::new(rx)
}
